package com.xuzz.study.spark.java;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;


/**
 * 简单的stream
 * 从本地服务器的8080端口接收一行一行的数据,并且筛选出包含error的行,打印
 */
public class Stream1 {
    public static void main(String[] args)
    {
        SparkConf sparkConf = new SparkConf().setMaster("local[2]").setAppName("sparkStreaming");
        JavaStreamingContext javaStreamingContext=new JavaStreamingContext(sparkConf, Durations.seconds(1));

        //创建一个链接到8080的DStream
        JavaReceiverInputDStream<String> dStream = javaStreamingContext.socketTextStream("localhost",8080);
        JavaDStream<String> errorLine = dStream.filter((Function<String,Boolean>) s->s.contains("error"));

        //打印包含error的行
        errorLine.print();

        try
        {
            javaStreamingContext.start();
            javaStreamingContext.awaitTermination();
        }catch (InterruptedException e)
        {
            e.printStackTrace();
        }
    }
}
